package com.qianbao.finance.sync.modules.elasticsearch.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author：zgh
 * @DATE: 2018/6/27 16:26
 * @version: 0.0.1
 * @Description:
 */
@Component
public class CanalClientFactory  {
    private static final Logger logger = LoggerFactory.getLogger(CanalClientFactory.class);
    @Value("${canal.host}")
    private String canalHost;
    @Value("${canal.port}")
    private String canalPort;
    @Value("${canal.destination}")
    private String canalDestination;
    @Value("${canal.username}")
    private String canalUsername;
    @Value("${canal.password}")
    private String canalPassword;
    @Value("${canal.zkServer}")
    private String zkServers;
    private final  Map<String, CanalConnector> canalConnectorMap = new HashMap<>();

    @PostConstruct
    public void init() {
            String[] canalDestinations = canalDestination.split(",");
            for (String canalDestination : canalDestinations) {
                try{
                    CanalConnector canalConnector =
                            CanalConnectors.newClusterConnector(zkServers, canalDestination, "", "");
                    canalConnector.connect();
                    canalConnector.subscribe();
                    canalConnector.rollback();
                    canalConnectorMap.put(canalDestination, canalConnector);
                } catch (Exception e){
                    logger.error("创建"+canalDestination+"canal链接失败："+ JSON.toJSONString(e));

                }
            }
    }


    /**
     * 刷新某一渠道canal 链接
     * @param canalDestination
     */
    public void refresh(String canalDestination) {
            CanalConnector canalConnector =
                    CanalConnectors.newClusterConnector(zkServers, canalDestination, "", "");
            canalConnector.connect();
            canalConnector.subscribe();
            canalConnector.rollback();
            canalConnectorMap.put(canalDestination, canalConnector);

    }


    public Map<String, CanalConnector> get() {
        return canalConnectorMap;
    }

    public CanalConnector getCanalConnector(String canalDestination){

        return canalConnectorMap.get(canalDestination);
    }
}
